/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsrm.storage.impl;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Vector;

import cgl.narada.wsinfra.wsrm.storage.WsrmProtocolStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageEventsFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageException;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageWidget;

/**
 * <p>Title:WsrmProtocolStorageOperationsImpl </p>
 * <p>Description: Handles Protocol Storage Operations
 * which is defined in interface
 * WsrmProtocolStorageOperations</p>
 * @author Beytullah Yildiz
 */

public class WsrmProtocolStorageOperationsImpl
    implements WsrmProtocolStorageOperations {

  private WsrmQueryHandler wsrmQueryHandler;
  private String moduleName = "WsrmProtocolStorageOperationsImpl: ";

  public WsrmProtocolStorageOperationsImpl() {
  }

  public WsrmProtocolStorageOperationsImpl(Connection connection) {
    wsrmQueryHandler = new WsrmQueryHandler(connection);
  }

  /** Processes a set of acknowledgements received on a given sequence.
      This method throws an exception under the following conditions.
      <br> The Sequence specified in sequenceIdentifier is an unknown one.
      <br> One or more of the acknowledgements refer to non-existent message
      numbers.
      <br> Storage related problems */
  public void processAcknowledgementsOnSequence(String sequenceIdentifier,
                                                long[] ackMessageNumbers) throws
      WsrmStorageException {

    int size = ackMessageNumbers.length;
    for (int i = 0; i < size; i++) {
      processAcknowledgementOnSequence(sequenceIdentifier, ackMessageNumbers[i]);
    }

  }

  /** Processes an acknowledgement received on a given sequence for a specific
      message number.
      This method throws an exception under the following conditions.
      <br> The Sequence specified in sequenceIdentifier is an unknown one.
      <br> The acknowledgement refers to non-existent message number
      <br> Storage related problems */
  public void processAcknowledgementOnSequence(String sequenceIdentifier,
                                               long ackMessageNumber) throws
      WsrmStorageException {

    String query = null;
    query = "update protocollight  set ackinfo = " +
        wsrmQueryHandler.getIntOfBoolean(true) +
        " where sequenceid='" + sequenceIdentifier +
        "' and messagenumber=" + ackMessageNumber;

    int count = wsrmQueryHandler.updateQuery(query);
    System.out.println("count " + count);
    if (count <= 0) {
      throw new WsrmStorageException(moduleName +
          " processAcknowledgementsOnSequence exception [" + query + "] " +
          "sequence Identifier or message number does not exist");

    }

  }

  /** Retrieves the list of message numbers that have NOT been acknowledged,
      up until the message number specified in <i>upUntilMessageNumber</i>
      This method throws an exception under the following conditions.
      <br> The Sequence specified in sequenceIdentifier is an unknown one.
      <br> The specified message number is a non-existent one.
      <br> Storage related problems  */
  public long[] getUnacknowledgedMessageNumbers(String sequenceIdentifier,
                                                long upUntilMessageNumber) throws
      WsrmStorageException {


    String query = null;
    query = "select count(*) from protocollight where sequenceid='"
        + sequenceIdentifier + "'";

    if (!wsrmQueryHandler.isExistInTable(query)) {
      throw new WsrmStorageException(moduleName +
          " processAcknowledgementsOnSequence exception [" + query + "] " +
          "sequence Identifier does not exist");

    }

    query =
        "select count(*) from protocollight where messagenumber="
        + upUntilMessageNumber;

    if (!wsrmQueryHandler.isExistInTable(query)) {
      throw new WsrmStorageException(moduleName +
          " processAcknowledgementsOnSequence exception [" + query + "] " +
          "specified message number does not exist");
    }

    query =
        "select messagenumber from protocollight where sequenceid='"
        + sequenceIdentifier + "'" + "and ackinfo = 0"
        + " and messagenumber <= " + upUntilMessageNumber
        + " or timeinterval < " + System.currentTimeMillis(); ;

    PreparedStatement preparedStatement =
        wsrmQueryHandler.getPreparedStatement(query);
    ResultSet resultSet = wsrmQueryHandler.getResultSet(preparedStatement);

    long[] unacknowledgedMessageNumbers = wsrmQueryHandler.getLongNumbers(
        resultSet);

    wsrmQueryHandler.closeResultSet(resultSet);
    wsrmQueryHandler.closePreparedStatement(preparedStatement);

    return unacknowledgedMessageNumbers;

  }

  /** Retrieves the list of message numbers that have NOT been acknowledged
      for a given Sequence. This method throws an exception if the Sequence
      specified in sequenceIdentifier is an unknown one or if there are
      other storage problems. */
  public long[] getUnacknowledgedMessageNumbers(String sequenceIdentifier,
                                                boolean ackRequested) throws
      WsrmStorageException {

    String query =
        "select count(*) from protocollight where sequenceid='"
        + sequenceIdentifier + "'";

    if (!wsrmQueryHandler.isExistInTable(query)) {
      throw new WsrmStorageException(moduleName +
          " processAcknowledgementsOnSequence exception [" + query + "] " +
          "sequence Identifier does not exist");
    }

    if (ackRequested) {
      query =
          "select messagenumber from protocollight where sequenceid='" +
          sequenceIdentifier + "' and ackinfo =" +
          wsrmQueryHandler.getIntOfBoolean(false);
    }
    else {

      query =
          "select messagenumber from protocollight where sequenceid='" +
          sequenceIdentifier + "' and timeinterval < " +
          System.currentTimeMillis() + " and ackinfo = 0";

    }

    PreparedStatement preparedStatement =
        wsrmQueryHandler.getPreparedStatement(query);
    ResultSet resultSet = wsrmQueryHandler.getResultSet(preparedStatement);

    long[] unacknowledgedMessageNumbers = wsrmQueryHandler.getLongNumbers(
        resultSet);

    wsrmQueryHandler.closeResultSet(resultSet);
    wsrmQueryHandler.closePreparedStatement(preparedStatement);

    return unacknowledgedMessageNumbers;

  }

  /** Retrieves the list of message numbers that have been acknowledged
      This method throws WsrmStorageException.
   */
  public long[] getAcknowledgedMessageNumbers(String sequenceIdentifier) throws
      WsrmStorageException {

    String query = null;
    query = "select count(*) from protocollight where sequenceid='"
        + sequenceIdentifier + "'";

    if (!wsrmQueryHandler.isExistInTable(query)) {
      throw new WsrmStorageException(moduleName +
                                     " getAcknowledgedMessageNumbers exception"
                                     + "[" + query + "] " +
                                     "sequence Identifier does not exist");

    }

    query =
        "select messagenumber from protocollight where sequenceid='"
        + sequenceIdentifier + "'" + "and ackinfo = 1";

    PreparedStatement preparedStatement =
        wsrmQueryHandler.getPreparedStatement(query);
    ResultSet resultSet = wsrmQueryHandler.getResultSet(preparedStatement);

    long[] acknowledgedMessageNumbers =
        wsrmQueryHandler.getLongNumbers(resultSet);

    wsrmQueryHandler.closeResultSet(resultSet);
    wsrmQueryHandler.closePreparedStatement(preparedStatement);

    return acknowledgedMessageNumbers;

  }

  /** Retrieves the message number that was last assigned on a given
     sequence. This method throws an exception if the Sequence
     specified in sequenceIdentifier is an unknown one or if there are
     storage related problems.  */
  public long getMessageNumberLastAssigned(String sequenceIdentifier) throws
      WsrmStorageException {

    String query =
        "select max(messagenumber) from protocollight where sequenceid='"
        + sequenceIdentifier + "'";

    long messageNumber = wsrmQueryHandler.getLongNumber(query);

    return messageNumber;
  }

  /** Store a WSRM message to storage. This method throws an exception if
      there are storage related problems. */
  public void store(WsrmStorageWidget widget) throws WsrmStorageException {


    boolean ackInfo = widget.getAcknowledgementInfo();
    String sequenceId = widget.getSequenceIdentifier();
    long messageNumber = widget.getMessageNumber();
    byte[] protocolBinary = widget.getBytes();
    long timeInterval = widget.getAckTimestampInfo();

    String query =
        "insert into protocollight values('" + sequenceId + "',"
        + messageNumber + "," + timeInterval + ","+
        wsrmQueryHandler.getIntOfBoolean(ackInfo) + ")";

    PreparedStatement pstmt1 =
        wsrmQueryHandler.getPreparedStatement(query);
    try {
      //pstmt1.executeQuery();
      pstmt1.executeUpdate();
      pstmt1.close();
    }
    catch (SQLException ex) {
      throw new WsrmStorageException(moduleName +
                                     "store exception ["
                                     + query + "] " + ex.toString());
    }

    query = "insert into  protocolfull"
        + " values('" + sequenceId + "'," + messageNumber + ", ? )";
    PreparedStatement pstmt2 =
        wsrmQueryHandler.getPreparedStatement(query);
    try {
      pstmt2.setBytes(1, protocolBinary);
      //pstmt2.executeQuery();
      pstmt2.executeUpdate();
      pstmt2.close();

    }
    catch (SQLException ex) {
      throw new WsrmStorageException(moduleName +
                                     "store exception ["
                                     + query + "] " + ex.toString());
    }

    System.out.println(moduleName + "Protocol which has sequence id <" +
                       sequenceId + "> is stored");

  }

  private Vector getResultSetInVecotorForProtocol(ResultSet resultSet) throws
      WsrmStorageException {
    Vector resultVector = new Vector();

    try {
      while (resultSet.next()) {
        resultVector.add(resultSet.getBytes(1));
      }
    }
    catch (SQLException ex) {
      throw new WsrmStorageException(moduleName +
          " getResultSetInVecotorForSequenceInfo exception " + ex.toString());

    }

    return resultVector;
  }

  /** Checks to see if there are widgets that have been stored for the
      sequence in question. */
  public boolean hasStoredElements(String sequenceIdentifier) throws
      WsrmStorageException {
    boolean boolValue = false;
    if (sequenceIdentifier == null) {
      throw new WsrmStorageException(moduleName +
          " hasStoredElements exception sequence Identifier is null");

    }
    String query =
        "select count(*) from protocollight where sequenceid='"
        + sequenceIdentifier + "'";

    if (wsrmQueryHandler.isExistInTable(query)) {
      boolValue = true;
    }

    return boolValue;
  }

  /** Check to see if there is a widget that has been stored for a specific
      message number in a given Sequence */
  public boolean
      hasStoredElement(String sequenceIdentifier, long messageNumber) throws
      WsrmStorageException {

    boolean boolValue = false;
    if (sequenceIdentifier == null) {
      throw new WsrmStorageException(moduleName + "hasStoredElement() " +
                                     "exception sequence Identifier is null");

    }
    String query =
        "select count(*) from protocollight " +
        " where sequenceid ='" + sequenceIdentifier +
        "' and messagenumber =" + messageNumber;

    if (wsrmQueryHandler.isExistInTable(query)) {
      boolValue = true;
    }

    return boolValue;

  }

  /** Retrieves a stored WsrmStorageWidget from storage. This method throws an
      exception under the following conditions.
      <br> The Sequence specified in sequenceIdentifier is an unknown one.
      <br> The specified message number is a non-existent one.
      <br> Storage related problems  */
  public WsrmStorageWidget
      getStoredElement(String sequenceIdentifier,
                       long messageNumber) throws WsrmStorageException {
    String query = null;
    query =
        "select protocolbinary from protocolfull where sequenceid='"
        + sequenceIdentifier + "' and messagenumber=" + messageNumber;

    PreparedStatement preparedStatement =
        wsrmQueryHandler.getPreparedStatement(query);
    ResultSet resultSet = wsrmQueryHandler.getResultSet(preparedStatement);

    Vector byteArrayVector = getResultSetInVecotorForProtocol(resultSet);
    if (byteArrayVector.size() <= 0) {
      throw new WsrmStorageException(moduleName +
          " processAcknowledgementsOnSequence exception [" + query + "] " +
          "sequence identifier or specified message number do not exist");

    }
    byte[] byteArray = (byte[]) byteArrayVector.remove(0);

    WsrmStorageEventsFactory wsrmStorageEventsFactory =
        WsrmStorageEventsFactory.getInstance();
    WsrmStorageWidget wsrmStorageWidget = wsrmStorageEventsFactory.
        createWsrmStorageWidget(byteArray);

    wsrmQueryHandler.closeResultSet(resultSet);
    wsrmQueryHandler.closePreparedStatement(preparedStatement);
    return wsrmStorageWidget;
  }

  /** Retrieves stored WsrmStorageWidgets from storage. This method throws an
      exception under the following conditions.
      <br> The Sequence specified in sequenceIdentifier is an unknown one.
      <br> One or more of the specified message numbers are non-existent.
      <br> Storage related problems */
  public WsrmStorageWidget[]
      getStoredElements(String sequenceIdentifier,
                        long[] messageNumbers) throws WsrmStorageException {
    int size = messageNumbers.length;
    WsrmStorageWidget[] wsrmStorageWidget =
        new WsrmStorageWidget[size];
    for (int i = 0; i < size; i++) {
      wsrmStorageWidget[i] = getStoredElement(sequenceIdentifier,
                                              messageNumbers[i]);
    }

    return wsrmStorageWidget;
  }

  /** Reset the ack interval associated with the message */
  public void resetAckInterval(WsrmStorageWidget wsrmStorageWidget,
                               long ackInterval) throws WsrmStorageException {

    String sequenceId = wsrmStorageWidget.getSequenceIdentifier();
    long messageNumber = wsrmStorageWidget.getMessageNumber();
    byte[] protocolBinary = wsrmStorageWidget.getBytes();
    long timeInterval = ackInterval;

    String query =
        "update protocollight set timeinterval=" + timeInterval +
        " where sequenceid ='" + sequenceId + "' and messagenumber =" +
        messageNumber;

    PreparedStatement pstmt1 =
        wsrmQueryHandler.getPreparedStatement(query);
    try {
      pstmt1.executeUpdate();
      pstmt1.close();

    }
    catch (SQLException ex) {
      throw new WsrmStorageException(moduleName +
                                     "resetAckInterval exception ["
                                     + query + "] " + ex.toString());
    }

    query = "update protocolfull set protocolbinary= ? " +
        " where sequenceid ='" + sequenceId + "' and messagenumber =" +
        messageNumber;

    PreparedStatement pstmt2 =
        wsrmQueryHandler.getPreparedStatement(query);
    try {
      pstmt2.setBytes(1, protocolBinary);
      pstmt2.executeUpdate();
      pstmt2.close();

    }
    catch (SQLException ex) {
      throw new WsrmStorageException(moduleName +
                                     "resetAckInterval exception ["
                                     + query + "] " + ex.toString());
    }

  }

  public static void main(String[] args) {
    WsrmProtocolStorageOperationsImpl wsrmProtocolStorageOperationsImpl1 = new
        WsrmProtocolStorageOperationsImpl();

  }

}
